package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: 热门商品Top10
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark01_req {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")
    // (品类，点击数量）

    val clickCnt: RDD[(String, Int)] = userVisit
      .filter(
        line => {
          val datas: Array[String] = line.split("_")
          datas(6) != "-1"
        }
      )
      .map(
        line => {
          val datas: Array[String] = line.split("_")
          (datas(6), 1)
        }
      )
      .reduceByKey(_ + _)

    //统计下单的点击数量：（品类ID，下单数量）
    val orderCnt: RDD[(String, Int)] = userVisit
      .filter(
        line => {
          val datas: Array[String] = line.split("_")
          datas(8) != "null"
        }
      )
      .flatMap(
        line => {
          val datas: Array[String] = line.split("_")
          val cids: Array[String] = datas(8).split(",")
          cids.map((_, 1))
        }
      )
      .reduceByKey(_ + _)

    //统计支付的点击数量：（品类ID，支付数量）
    val payCnt: RDD[(String, Int)] = userVisit
      .filter(
        line => {
          val datas: Array[String] = line.split("_")
          datas(10) != "null"
        }
      )
      .flatMap(
        line => {
          val datas: Array[String] = line.split("_")
          val cids: Array[String] = datas(10).split(",")
          cids.map((_, 1))
        }
      )
      .reduceByKey(_ + _)


    clickCnt.cogroup(orderCnt,payCnt)
      .mapValues{
        case (clickIter,orderIter,payIter)=>{
          var orderCnt = 0;
          var payCnt = 0;
          var clickCnt = 0;
          if (clickIter.iterator.hasNext){
            clickCnt = clickIter.iterator.next()
          }
          if (orderIter.iterator.hasNext){
            orderCnt = orderIter.iterator.next()
          }

          if (payIter.iterator.hasNext){
            payCnt = payIter.iterator.next()
          }

          (clickCnt,orderCnt,payCnt)
        }
      }.sortBy(_._2,false).take(10).foreach(println)


    sc.stop()
  }
}
